package cdrprocessing.bolt.rule;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import cdrprocessing.bolt.util.WindowAwareBolt;
import cdrprocessing.common.CallPeriod;
import cdrprocessing.common.rules.AbstractRule;
import cdrprocessing.common.rules.DiscountForFailureRule;
import cdrprocessing.common.rules.DiscountForManyCallsRule;
import cdrprocessing.common.rules.DiscountForPopularUsersRule;

public class RuleEvaluatorBolt extends WindowAwareBolt {

    private List<AbstractRule> rules;

    private Map<Integer, AreaFailureCount> areaFailureCounts;

    private Map<Long, CallSummary> callSummaries;

    private SimpleDateFormat df;

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("DiscountForPopularUsersRule", new Fields("phoneNumber", "rule"));
        declarer.declareStream("DiscountForManyCallsRule", new Fields("phoneNumber", "rule"));
        declarer.declareStream("DiscountForFailureRule", new Fields("areaCode", "rule"));
    }

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        super.prepare(stormConf, context, collector);
        df = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
        rules = new ArrayList<AbstractRule>();
        areaFailureCounts = new HashMap<Integer, AreaFailureCount>();
        callSummaries = new HashMap<Long, CallSummary>();
    }

    @Override
    protected void handleStartWindow() {
        callSummaries.clear();
        areaFailureCounts.clear();
    }

    @Override
    protected void handleEndWindow() {
    }

    @Override
    protected void handleTuple(Tuple input) {
        try {
            if (input.getSourceStreamId().equals("CdrStream")) {
                processAreaFailure(input);

                processCallSummary(input);

                processEvent(input);
            } else if (input.getSourceStreamId().startsWith("RuleStream")) {
                final String action = input.getStringByField("action");
                if ("activate".equals(action)) {
                    rules.add((AbstractRule) input.getValueByField("rule"));
                } else {
                    rules.remove((AbstractRule) input.getValueByField("rule"));
                }
            } else {
                System.out.println("Unexpected tuple " + input);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private void processAreaFailure(Tuple input) {
        final int terminationReason = input.getIntegerByField("terminateReason");
        if (terminationReason == 3 || terminationReason == 4) {
            final int area = input.getIntegerByField("area");
            if (!areaFailureCounts.containsKey(area)) {
                areaFailureCounts.put(area, new AreaFailureCount());
            }

            areaFailureCounts.get(area).incrementFaillureCount();
        }
    }

    private void processCallSummary(Tuple input) throws ParseException {
        final long callingNumber = input.getLongByField("customer");
        if (!callSummaries.containsKey(callingNumber)) {
            callSummaries.put(callingNumber, new CallSummary());
        }

        final CallSummary summary = callSummaries.get(callingNumber);

        summary.addDuration(input.getDoubleByField("duration"));

        final CallPeriod calledTime = (CallPeriod) input.getValueByField("period");
        if (calledTime == CallPeriod.MORNING) {
            summary.incrementMorningCount();
        } else if (calledTime == CallPeriod.AFTERNOON) {
            summary.incrementAfternoonCount();
        } else {
            summary.incrementEveningCount();
        }
    }

    private void processEvent(Tuple input) {
        for (final AbstractRule rule : rules) {
            if ("DiscountForPopularUsersRule".equals(rule.getType())) {
                handleDiscountForPopularUsersRule((DiscountForPopularUsersRule) rule, input.getLongByField("customer"));
            } else if ("DiscountForManyCallsRule".equals(rule.getType())) {
                handleDiscountForManyCallsRule((DiscountForManyCallsRule) rule, input.getLongByField("customer"));
            } else if ("DiscountForFailureRule".equals(rule.getType())) {
                handleDiscountForFailureRule((DiscountForFailureRule) rule, input.getIntegerByField("area"));
            } else {
                System.out.println("Unexpected rule " + rule);
            }
        }
    }

    private void handleDiscountForPopularUsersRule(DiscountForPopularUsersRule rule, long callingNumber) {
        final CallSummary summary = callSummaries.get(callingNumber);
        if (summary == null || summary.hasRule(rule.getId())) {
            return;
        }

        final long timeGoal = rule.getTimeAmount() * ("hours".equals(rule.getTimeUnit()) ? 60 * 60 : 60);
        if (summary.getDuration() >= timeGoal) {
            summary.addRuleFired(rule.getId());
            collector.emit("DiscountForPopularUsersRule", new Values(callingNumber, rule));
        }
    }

    private void handleDiscountForManyCallsRule(DiscountForManyCallsRule rule, long callingNumber) {
        final CallSummary summary = callSummaries.get(callingNumber);
        if (summary == null || summary.hasRule(rule.getId())) {
            return;
        }

        final int callCount = "morning".equals(rule.getPeriod()) ? summary.getMorningCount() : "afternoon".equals(rule.getPeriod()) ? summary
                .getAfternoonCount() : summary.getEveningCount();
        if (callCount >= rule.getNumberOfCalls()) {
            summary.addRuleFired(rule.getId());
            collector.emit("DiscountForManyCallsRule", new Values(callingNumber, rule));
        }
    }

    private void handleDiscountForFailureRule(DiscountForFailureRule rule, long areaNumber) {
        final AreaFailureCount failureCount = areaFailureCounts.get(areaNumber);
        if (failureCount == null || failureCount.hasRule(rule.getId())) {
            return;
        }

        if (failureCount.getFailureCount() >= rule.getNumberOfCalls()) {
            failureCount.addRuleFired(rule.getId());
            collector.emit("DiscountForFailureRule", new Values(areaNumber, rule));
        }
    }

    private final class CallSummary {

        private double duration = 0;

        private int morningCount = 0;

        private int afternoonCount = 0;

        private int eveningCount = 0;

        private Set<Integer> rulesFired = new HashSet<Integer>();

        public double getDuration() {
            return duration;
        }

        public int getMorningCount() {
            return morningCount;
        }

        public int getAfternoonCount() {
            return afternoonCount;
        }

        public int getEveningCount() {
            return eveningCount;
        }

        public void incrementMorningCount() {
            morningCount++;
        }

        public void incrementAfternoonCount() {
            afternoonCount++;
        }

        public void incrementEveningCount() {
            eveningCount++;
        }

        public void addDuration(double value) {
            duration += value;
        }

        public void addRuleFired(int ruleId) {
            rulesFired.add(ruleId);
        }

        public boolean hasRule(int ruleId) {
            return rulesFired.contains(ruleId);
        }
    }

    private final class AreaFailureCount {

        private Set<Integer> rulesFired = new HashSet<Integer>();

        private int failureCount = 0;

        public void incrementFaillureCount() {
            failureCount++;
        }

        public int getFailureCount() {
            return failureCount;
        }

        public void addRuleFired(int ruleId) {
            rulesFired.add(ruleId);
        }

        public boolean hasRule(int ruleId) {
            return rulesFired.contains(ruleId);
        }

    }
}
